package com.xuliugen.storm.demo.simple.bolts;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

public class UpperBolt extends BaseRichBolt {

    private static final long serialVersionUID = 1L;

    private OutputCollector collector;

    @Override
    public void execute(Tuple tuple) {
        // 先获取到上一个组件传递过来的数据,数据在tuple里面
        String lowerName = tuple.getString(0);

        // 将姓名转换成大写
        String upperName = lowerName.toUpperCase();

        // 将转换完成的商品名发送出去
        collector.emit(new Values(upperName));

    }

    @Override
    public void prepare(Map conf, TopologyContext topologyContext, OutputCollector collector) {
        this.collector = collector;
    }

    /**
     * 声明该bolt组件要发出去的tuple的字段
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("upperName"));
    }

}
